package org.apache.rocketmq.consumer;

import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * @Description 主动拉消息模式
 * @Author lijing1@wxchina.com
 * @Date 2021/2/4 13:29
 **/
public class PullConsumer {

    public static final java.lang.String NAME_SRV_ADDR = "localhost:9876";

    public static volatile boolean running = true;


    public static void main(String[] args) throws MQClientException {
        DefaultLitePullConsumer litePullConsumer = new DefaultLitePullConsumer(
                "lite_pull_consumer_test1");
        litePullConsumer.setNamesrvAddr(NAME_SRV_ADDR);
        litePullConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        litePullConsumer.subscribe("TopicTest", "tag1 || tag2");
        litePullConsumer.start();
        litePullConsumer.setAutoCommit(false);
        loopMsg(litePullConsumer);

        /**

         */


    }

    private static void loopMsg(DefaultLitePullConsumer litePullConsumer) {
        try {
            while (running) {
                List<MessageExt> messageExts = litePullConsumer.poll();
                for (MessageExt messageExt : messageExts) {
                    System.out.println(messageExt);
                    System.out.printf("%s%n", new java.lang.String(messageExt.getBody()));
                    litePullConsumer.commitSync();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            litePullConsumer.shutdown();
        }

    }
}
